package cn.itcast.flink.table;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;

/**
 * Author itcast
 * Date 2021/7/30 6:55
 * Desc TODO
 */
public class FlinkSQLTableDemo {
    public static void main(String[] args) throws Exception {
        //1.准备环境 创建流环境 和 流表环境，并行度设置为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //2.Source 创建数据集
        //2.Source
        DataStream<Order> orderA = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(1L, "diaper", 4),
                new Order(3L, "rubber", 2)));

        DataStream<Order> orderB = env.fromCollection(Arrays.asList(
                new Order(2L, "pen", 3),
                new Order(2L, "rubber", 3),
                new Order(4L, "beer", 1)));

        //3.注册表 将数据流转换成表
        Table orders = tEnv.fromDataStream(orderA);
        // 通过fromDataStream将数据流转换成表
        // 将数据流转换成 创建临时视图
        tEnv.createTemporaryView("t_orderB",orderB);
        //4.执行查询，查询order1的amount>2并union all 上 order2的amoun<2的数据生成表
        Table table = tEnv.sqlQuery("select user,product,amount " +
                "from " + orders +
                " where amount>2 union all " +
                " select user,product,amount " +
                "from t_orderB " +
                " where amount<2");
        //4.1 将结果表转换成toAppendStream数据流
        DataStream<Tuple2<Boolean, Row>> result = tEnv.toRetractStream(table, Row.class);
        //5.打印结果
        result.print();
        //6.执行环境
        env.execute();
        // 创建实体类 user:Long product:String amount:int
    }
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Order {
        public Long user;
        public String product;
        public int amount;
    }
}
